[SPARK-38909][BUILD][CORE][YARN][FOLLOWUP] Make some code cleanup related to shuffle state db #37648

LuciferYang · 2022-08-24T13:23:43Z

What changes were proposed in this pull request?

This is a followup of #36200, the main change of this pr as follows:

Make DB and DBProvider as @Private to make the API intent clearer
Delete remove method from DBProvider witch is unnecessary @Override
Remove the useless null check condition from ExternalShuffleBlockResolver and RemoteBlockPushResolver, fix related sutes
Correction log print content

Why are the changes needed?

Fix new comments after #36200 merged.

Does this PR introduce any user-facing change?

No

How was this patch tested?

Pass All GitHub Actions

LuciferYang · 2022-08-24T13:26:26Z

cc @Ngone51 this pr used to fix the comments you left in #36200,

common/network-common/pom.xml

This reverts commit 2caffd1.

Ngone51 · 2022-08-26T05:56:51Z

...ork-shuffle/src/main/java/org/apache/spark/network/shuffle/ExternalShuffleBlockResolver.java

      .build(indexCacheLoader);
-    DBBackend dbBackend = null;
+    DBBackend dbBackend;
    if (registeredExecutorFile != null) {


Do we really need this condition? I feel like it's useless.

hmm... registeredExecutorFile may be null, for example, the following code path:

spark/core/src/main/scala/org/apache/spark/deploy/ExternalShuffleService.scala

Lines 81 to 92 in 68c47d5

protected def newShuffleBlockHandler(conf: TransportConf): ExternalBlockHandler = {

if (sparkConf.get(config.SHUFFLE_SERVICE_DB_ENABLED) && enabled) {

val shuffleDBName = sparkConf.get(config.SHUFFLE_SERVICE_DB_BACKEND)

val dBBackend = DBBackend.byName(shuffleDBName)

logInfo(s"Configured ${config.SHUFFLE_SERVICE_DB_BACKEND.key} as $shuffleDBName " +

s"and actually used value ${dBBackend.name()} ")

new ExternalBlockHandler(conf,

findRegisteredExecutorsDBFile(dBBackend.fileName(registeredExecutorsDB)))

} else {

new ExternalBlockHandler(conf, null)

}

}

spark/common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/ExternalBlockHandler.java

Lines 79 to 84 in 68c47d5

public ExternalBlockHandler(TransportConf conf, File registeredExecutorFile)

throws IOException {

this(new OneForOneStreamManager(),

new ExternalShuffleBlockResolver(conf, registeredExecutorFile),

new NoOpMergedShuffleFileManager(conf, null));

}

However, without adding this condition, the result will not change

EDIT ~~And the log will look strange without this condition~~

e3500cf remove the condition

This reverts commit 86681e3.

mridulm · 2022-08-26T18:58:23Z

...ork-shuffle/src/main/java/org/apache/spark/network/shuffle/ExternalShuffleBlockResolver.java

+    String dbBackendName =
+      conf.get(Constants.SHUFFLE_SERVICE_DB_BACKEND, DBBackend.LEVELDB.name());
+    DBBackend dbBackend = DBBackend.byName(dbBackendName);
+    db = DBProvider.initDB(dbBackend, this.registeredExecutorFile, CURRENT_VERSION, mapper);


Review note: Recovery need not be enabled for node managers - in which case registeredExecutorFile will be null (in addition to tests).

DBProvider.initDB does handle null input though.

So the main change in this file and RemoteBlockPushResolver is moving the log message into if (db != null)

mridulm · 2022-08-26T19:02:37Z

Is this still WIP @LuciferYang ?

LuciferYang · 2022-08-27T02:10:33Z

Removed [WIP], and need @Ngone51 confirm whether there are new comments of #36200.

And I have a point to discuss:

There are two initLevelDB method in LevelDBProvider, one of them is only for testing code and only used by org.apache.spark.network.shuffle.ShuffleTestAccessor now, so should we move the test only initLevelDB method to ShuffleTestAccessor? The implementation of RocksDB will also encounter this issue

LuciferYang · 2022-08-29T03:15:48Z

Removed [WIP], and need @Ngone51 confirm whether there are new comments of #36200.

And I have a point to discuss:

There are two initLevelDB method in LevelDBProvider, one of them is only for testing code and only used by org.apache.spark.network.shuffle.ShuffleTestAccessor now, so should we move the test only initLevelDB method to ShuffleTestAccessor? The implementation of RocksDB will also encounter this issue

@Ngone51 @mridulm 7a22aeb Move initDB method used only by test back to ShuffleTestAccessor

LuciferYang · 2022-08-31T07:18:03Z

Do you have time to further review this one? Thanks @mridulm

mridulm

Just a couple of comments - but since this is to mostly address @Ngone51's feedback; would be good if you can review it !

common/network-common/pom.xml

common/network-common/src/main/java/org/apache/spark/network/util/LevelDBProvider.java

mridulm · 2022-08-31T08:21:33Z

...urce-managers/yarn/src/test/scala/org/apache/spark/network/shuffle/ShuffleTestAccessor.scala

+      }
+    } else {
+      null
+    }


We will end up duplicating logic between DBProvider and ShuffleTestAccessor - let us consolidate it in DBProvider.
This also impacts the integration test anyway

OK, let me revert this change first and then see if there is a better way

do some check, I think we can directly use another initDB and delete this one.

This reverts commit 7a22aeb.

Ngone51 · 2022-09-02T12:22:15Z

...rce-managers/yarn/src/test/scala/org/apache/spark/network/yarn/YarnShuffleServiceSuite.scala

 import org.apache.spark.network.yarn.util.HadoopConfigProvider
 import org.apache.spark.tags.ExtendedLevelDBTest
 import org.apache.spark.util.Utils
-


revert this?

LuciferYang · 2022-09-04T03:14:56Z

should we merge this one ? @Ngone51 @mridulm

mridulm · 2022-09-06T05:40:12Z

...urce-managers/yarn/src/test/scala/org/apache/spark/network/shuffle/ShuffleTestAccessor.scala

+      version: StoreVersion,
+      mapper: ObjectMapper)
+    : ConcurrentMap[ExternalShuffleBlockResolver.AppExecId, ExecutorShuffleInfo] = {
+    val db = DBProvider.initDB(dbBackend, file, version, mapper)


This is a behavior change for test, right (this ?)
Essentially, YarnShuffleIntegrationSuite would expect it to fail with an exception and not do error handling (which initDB does).

spark/resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/YarnShuffleIntegrationSuite.scala

Lines 155 to 170 in 19b1780

try {

val data = sc.parallelize(0 until 100, 10).map { x => (x % 10) -> x }.reduceByKey{ _ + _ }.

collect().toSet

sc.listenerBus.waitUntilEmpty(WAIT_TIMEOUT_MILLIS)

data should be ((0 until 10).map{x => x -> (x * 10 + 450)}.toSet)

result = "success"

// only one process can open a leveldb file at a time, so we copy the files

if (registeredExecFile != null && execStateCopy != null) {

val dbBackendName = conf.get(SHUFFLE_SERVICE_DB_BACKEND.key)

val dbBackend = DBBackend.byName(dbBackendName)

logWarning(s"Configured ${SHUFFLE_SERVICE_DB_BACKEND.key} as $dbBackendName " +

s"and actually used value ${dbBackend.name()}")

FileUtils.copyDirectory(registeredExecFile, execStateCopy)

assert(!ShuffleTestAccessor

.reloadRegisteredExecutors(dbBackend, execStateCopy).isEmpty)

}

Do you mean the behavior change is execStateCopy exists but open failed? The original method will throw an exception directly, and the new method will return a new instance? It seems that the case just to test whether an existing execStateCopy can be loaded again?

If the original case does not test the execStateCopy exists but open failed scenario, it seems that there is no change.

For the initDB(dbBackend, file) method:

spark/common/network-common/src/main/java/org/apache/spark/network/util/LevelDBProvider.java

Lines 88 to 94 in 19b1780

@VisibleForTesting

static DB initLevelDB(File file) throws IOException {

Options options = new Options();

options.createIfMissing(true);

JniDBFactory factory = new JniDBFactory();

return factory.open(file, options);

}

if execStateCopy exists, it will re-open it and load data from execStateCopy

else if execStateCopy not exists, it will open a new db and return empty data due to options.createIfMissing(true), then assert failed

For the initDB(dbBackend, dbFile, version, mapper) method:

if execStateCopy exists, it will re-open it and load data from execStateCopy

else if execStateCopy not exists, it will open a new db and return empty data, then assert failed

For execStateCopy exists but open failed scenario, use initDB(dbBackend, file) will throw an unhandled exception, use initDB(dbBackend, dbFile, version, mapper) will assert failed. Is this unacceptable?

@mridulm any suggestions?

@squito or @tgravescs should have more context about this, and comment better - looking at the test, I would expect this test code path to additionally test if the DB is missing/invalid/etc and fail in case there is any issues (due to lack of error handling/fallback in LevelDBProvider.initLevelDB) - which changes with in PR, that no longer happens; right ?

@mridulm add SPARK-40364 to tracking this

I will give a separately pr to continue to explore the feasibility about this change

sorry didn't get a chance to look before your filed the other issue, happy to look if you make the change under the other issue

Thanks @tgravescs , I will give another pr later ~

This reverts commit ec370e7.

LuciferYang · 2022-09-07T06:43:36Z

Run / Base image build failed as follows:

2022-09-07T04:35:50.8451091Z #32 exporting to image
2022-09-07T04:35:50.8451363Z #32 exporting layers done
2022-09-07T04:35:50.8452103Z #32 exporting manifest sha256:1738df8ea1b7b1d839a5fdaa47f4a3d414d44c331688eddf60df5e9529e1fad5 done
2022-09-07T04:35:50.8452943Z #32 exporting config sha256:9e4187b41de14caaad1f643e02cb8ac50ec0dbb65d3dea2dfa4397b86addfb1d
2022-09-07T04:35:50.9955812Z #32 exporting config sha256:9e4187b41de14caaad1f643e02cb8ac50ec0dbb65d3dea2dfa4397b86addfb1d done
2022-09-07T04:35:50.9956540Z #32 pushing layers
2022-09-07T04:35:51.2964868Z #32 ...
2022-09-07T04:35:51.2965252Z 
2022-09-07T04:35:51.2966013Z #33 [auth] luciferyang/apache-spark-ci-image:pull,push token for ghcr.io
2022-09-07T04:35:51.2966363Z #33 DONE 0.0s
2022-09-07T04:35:51.4476109Z 
2022-09-07T04:35:51.4479341Z #32 exporting to image
2022-09-07T04:35:51.8931704Z #32 ...
2022-09-07T04:35:51.8932307Z 
2022-09-07T04:35:51.8933413Z #34 [auth] apache/spark/apache-spark-github-action-image-cache:pull luciferyang/apache-spark-ci-image:pull,push token for ghcr.io
2022-09-07T04:35:51.8934022Z #34 DONE 0.0s
2022-09-07T04:35:52.0437234Z 
2022-09-07T04:35:52.0437900Z #32 exporting to image
2022-09-07T04:35:52.4041282Z #32 pushing layers 1.5s done
2022-09-07T04:35:52.4041635Z #32 ERROR: unexpected status: 403 Forbidden
2022-09-07T04:35:52.4042165Z ------
2022-09-07T04:35:52.4042409Z  > exporting to image:
2022-09-07T04:35:52.4042677Z ------
2022-09-07T04:35:52.4081024Z ERROR: failed to solve: unexpected status: 403 Forbidden
2022-09-07T04:35:52.4215414Z ##[error]buildx failed with: ERROR: failed to solve: unexpected status: 403 Forbidden

friendly ping @Yikun for help, I should not have changed GitHub's configuration

Yikun · 2022-09-07T06:49:21Z

pls rebase this PR, after #37815 merged.

Related github incident: https://www.githubstatus.com/incidents/d181frs643d4

LuciferYang · 2022-09-07T07:00:41Z

pls rebase this PR, after #37815 merged.

Thanks ~

Yikun · 2022-09-07T09:18:48Z

You can retrigger the ci, the issue github already fix the https://www.githubstatus.com/incidents/d181frs643d4

mridulm · 2022-09-08T01:47:33Z

Merged to master.
Thanks for working on this @LuciferYang !
Thanks for flagging this issue and the review @Ngone51 :-)

LuciferYang · 2022-09-08T02:39:27Z

Thanks for your review @mridulm @tgravescs ~
Thanks for helping fix GA @Yikun ~

fix wuyi's comments

579956e

LuciferYang marked this pull request as draft August 24, 2022 13:23

LuciferYang changed the title ~~[SPARK-38909][CORE][YARN][FOLLOWUP] Name later~~ [WIP][SPARK-38909][CORE][YARN][FOLLOWUP] Name later Aug 24, 2022

github-actions bot added CORE YARN labels Aug 24, 2022

LuciferYang changed the title ~~[WIP][SPARK-38909][CORE][YARN][FOLLOWUP] Name later~~ [WIP][SPARK-38909][CORE][YARN][FOLLOWUP] Aug 24, 2022

add @DeveloperAPI annotation

2caffd1

github-actions bot added the BUILD label Aug 25, 2022

LuciferYang commented Aug 25, 2022

View reviewed changes

common/network-common/pom.xml Show resolved Hide resolved

LuciferYang changed the title ~~[WIP][SPARK-38909][CORE][YARN][FOLLOWUP]~~ [WIP][SPARK-38909][BUILD][CORE][YARN][FOLLOWUP] Aug 25, 2022

Revert "add @DeveloperAPI annotation"

87c42dd

This reverts commit 2caffd1.

Ngone51 reviewed Aug 26, 2022

View reviewed changes

LuciferYang added 5 commits August 26, 2022 14:31

remove condition

e3500cf

part revert

86681e3

Revert "part revert"

7e3befa

This reverts commit 86681e3.

fix test

6ceb2d0

delete remove from LevelDBIteraot

11b9dad

mridulm reviewed Aug 26, 2022

View reviewed changes

mridulm mentioned this pull request Aug 26, 2022

[SPARK-38909][CORE][YARN] Encapsulate LevelDB used to store remote/external shuffle state as DB #36200

Closed

make db and dbiterator private

8b77b61

LuciferYang changed the title ~~[WIP][SPARK-38909][BUILD][CORE][YARN][FOLLOWUP]~~ [WIP][SPARK-38909][BUILD][CORE][YARN][FOLLOWUP] Make some code cleanup related to shuffle state db Aug 27, 2022

LuciferYang marked this pull request as ready for review August 27, 2022 02:05

LuciferYang changed the title ~~[WIP][SPARK-38909][BUILD][CORE][YARN][FOLLOWUP] Make some code cleanup related to shuffle state db~~ [SPARK-38909][BUILD][CORE][YARN][FOLLOWUP] Make some code cleanup related to shuffle state db Aug 27, 2022

LuciferYang added 2 commits August 29, 2022 11:04

Merge branch 'upmaster' into SPARK-38909-followup

3e59b7b

move test only initDB back to ShuffleTestAccessor

7a22aeb

Merge branch 'upmaster' into SPARK-38909-followup

9449457

mridulm reviewed Aug 31, 2022

View reviewed changes

common/network-common/pom.xml Show resolved Hide resolved

common/network-common/src/main/java/org/apache/spark/network/util/LevelDBProvider.java Show resolved Hide resolved

mridulm reviewed Aug 31, 2022

View reviewed changes

LuciferYang added 3 commits August 31, 2022 17:10

Revert "move test only initDB back to ShuffleTestAccessor"

1f3b62b

This reverts commit 7a22aeb.

use defualt initDB

ec370e7

imports

11850af

LuciferYang mentioned this pull request Aug 31, 2022

[SPARK-38888][BUILD][CORE][YARN][DOCS] Add RocksDB support for shuffle service state store #37610

Closed

Merge branch 'upmaster' into SPARK-38909-followup

024ec63

Ngone51 reviewed Sep 2, 2022

View reviewed changes

Ngone51 approved these changes Sep 2, 2022

View reviewed changes

revert blank line

c8ac7e4

Merge branch 'upmaster' into SPARK-38909-followup

33c96a6

mridulm reviewed Sep 6, 2022

View reviewed changes

LuciferYang added 4 commits September 7, 2022 10:41

Revert "use defualt initDB"

a5df1c4

This reverts commit ec370e7.

clean up improts

dfe7ce5

clean up improts

6c91380

Merge branch 'upmaster' into SPARK-38909-followup

8f47c15

mridulm closed this in 3566657 Sep 8, 2022

LuciferYang mentioned this pull request Sep 8, 2022

[SPARK-40364][CORE] Use the unified DBProvider#initDB method #37826

Closed

	protected def newShuffleBlockHandler(conf: TransportConf): ExternalBlockHandler = {
	if (sparkConf.get(config.SHUFFLE_SERVICE_DB_ENABLED) && enabled) {
	val shuffleDBName = sparkConf.get(config.SHUFFLE_SERVICE_DB_BACKEND)
	val dBBackend = DBBackend.byName(shuffleDBName)
	logInfo(s"Configured ${config.SHUFFLE_SERVICE_DB_BACKEND.key} as $shuffleDBName " +
	s"and actually used value ${dBBackend.name()} ")
	new ExternalBlockHandler(conf,
	findRegisteredExecutorsDBFile(dBBackend.fileName(registeredExecutorsDB)))
	} else {
	new ExternalBlockHandler(conf, null)
	}
	}

	public ExternalBlockHandler(TransportConf conf, File registeredExecutorFile)
	throws IOException {
	this(new OneForOneStreamManager(),
	new ExternalShuffleBlockResolver(conf, registeredExecutorFile),
	new NoOpMergedShuffleFileManager(conf, null));
	}

	try {
	val data = sc.parallelize(0 until 100, 10).map { x => (x % 10) -> x }.reduceByKey{ _ + _ }.
	collect().toSet
	sc.listenerBus.waitUntilEmpty(WAIT_TIMEOUT_MILLIS)
	data should be ((0 until 10).map{x => x -> (x * 10 + 450)}.toSet)
	result = "success"
	// only one process can open a leveldb file at a time, so we copy the files
	if (registeredExecFile != null && execStateCopy != null) {
	val dbBackendName = conf.get(SHUFFLE_SERVICE_DB_BACKEND.key)
	val dbBackend = DBBackend.byName(dbBackendName)
	logWarning(s"Configured ${SHUFFLE_SERVICE_DB_BACKEND.key} as $dbBackendName " +
	s"and actually used value ${dbBackend.name()}")
	FileUtils.copyDirectory(registeredExecFile, execStateCopy)
	assert(!ShuffleTestAccessor
	.reloadRegisteredExecutors(dbBackend, execStateCopy).isEmpty)
	}

	@VisibleForTesting
	static DB initLevelDB(File file) throws IOException {
	Options options = new Options();
	options.createIfMissing(true);
	JniDBFactory factory = new JniDBFactory();
	return factory.open(file, options);
	}

[SPARK-38909][BUILD][CORE][YARN][FOLLOWUP] Make some code cleanup related to shuffle state db #37648

[SPARK-38909][BUILD][CORE][YARN][FOLLOWUP] Make some code cleanup related to shuffle state db #37648

Uh oh!

Conversation

LuciferYang commented Aug 24, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

Uh oh!

LuciferYang commented Aug 24, 2022

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

LuciferYang Aug 26, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mridulm Aug 26, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mridulm commented Aug 26, 2022

Uh oh!

LuciferYang commented Aug 27, 2022

Uh oh!

LuciferYang commented Aug 29, 2022

Uh oh!

LuciferYang commented Aug 31, 2022

Uh oh!

mridulm left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

LuciferYang commented Sep 4, 2022

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

LuciferYang Sep 6, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

LuciferYang commented Sep 7, 2022

Uh oh!

Yikun commented Sep 7, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

LuciferYang commented Aug 24, 2022 •

edited

Loading

LuciferYang Aug 26, 2022 •

edited

Loading

mridulm Aug 26, 2022 •

edited

Loading

LuciferYang Sep 6, 2022 •

edited

Loading

Yikun commented Sep 7, 2022 •

edited

Loading